package com.sunlands.deskmate.message.handler.mq;

import com.alibaba.fastjson.JSONObject;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.channel.ChannelManager;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEvent;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEventEnum;
import com.sunlands.deskmate.message.handler.sender.trigger.SendMessageTrigger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.io.IOException;
import java.util.Objects;
import static com.sunlands.deskmate.config.RabbitMqConfig.EVENT_QUEUE_LOOKUP;

/**
 * 从storage转过来的消息, 推送给网页端
 *
 * @author anjl
 */
@Slf4j
@Component
public class LookupMessageWorker extends AbstractInboundMessageWorker {

    public static final String LOOKUP = "lookup";

    @Value("${WORKER_ID}")
    String workerId;

    @Autowired
    private SendMessageTrigger sendDispatcher;

    /**
     * 消费消息
     * <p>
     * 多节点情况下, 定义多个queue对用多个消费者, 使用fanout, 对等消费, 每个节点只消费和自己有关的
     *
     * @param mqMessage Message
     * @param channel   channel
     */
    @RabbitListener(queues = EVENT_QUEUE_LOOKUP + "${WORKER_ID}")
    public void run(Message mqMessage, com.rabbitmq.client.Channel channel) {
        String message = new String(mqMessage.getBody());

        log.info("lookup queue {}", message);

        Messages msgEntity;
        try {
            if (StringUtils.isEmpty(message)) {
                log.warn("messages is empty");
                handle(true, mqMessage, channel);
                return;
            }

            msgEntity = JSONObject.parseObject(message, Messages.class);
            checkCause(checkMsg(msgEntity), message, mqMessage, channel);
        } catch (Exception e) {
            log.error("messages invalid {}", message);
            try {
                handle(true, mqMessage, channel);
            } catch (IOException e1) {
                e1.printStackTrace();
            }
            return;
        }

        boolean success = false;
        try {
            String msgType = msgEntity.getType();
            long msgId = msgEntity.getAckId();
            Long toId = msgEntity.getRealToId();

            ChannelInfo channelInfo = new ChannelInfo(toId);
            if (checkChannel(channelInfo.getSessionKey())) {

                //消息派送事件
                TriggerEvent dispatchEvent = new TriggerEvent(TriggerEventEnum.UNSYNC, channelInfo, msgId, msgType, null);
                success = sendDispatcher.offer(dispatchEvent);
                log.debug("sendDispatcher.offer channelInfo={}, messages={}, result={}", channelInfo, msgEntity, success);
                handle(success, mqMessage, channel);
            } else {
                //忽略消息
                handle(true, mqMessage, channel);
            }
        } catch (IOException e) {
            try {
                log.warn("lookeup IOException success={} messages={}", success, message);
                handle(true, mqMessage, channel);
            } catch (Exception ex) {
                log.error("lookeup Exception success={} messages={}", success, message);
            }
        } catch (Exception e) {
            try {
                handle(true, mqMessage, channel);
            } catch (IOException ex) {
                log.error("lookeup IOException messages=" + message, ex);
            }
            log.error("lookeup Exception messages=" + message, e);
        }

    }

    private boolean checkChannel(String sessionKey) {
        return Objects.nonNull(ChannelManager.getChannel(sessionKey));
    }

}
